/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce;

import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.InvocationTargetException;
import java.io.IOException;
import org.sss.mapreduce.datatype.Packable;
import org.sss.mapreduce.datatype.SelfPackable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SelfEncoding implements Encoding {
  @SuppressWarnings("unused")
  private static Logger logger = LoggerFactory.getLogger(SelfEncoding.class);

  @Override
  public boolean canEncode(Class<? extends Packable> klass) throws SssException {
    try {
      klass.getConstructor();
    }
    catch (NoSuchMethodException e) {
      return false;
    }
    return SelfPackable.class.isAssignableFrom(klass);
  }

  @Override
  public Encoder createEncoder(Class<? extends Packable> klass) throws SssException {
    assert canEncode(klass): "invalid type";
    try {
      final Method method = klass.getMethod("createEncoder");
      if (method != null && Modifier.isStatic(method.getModifiers())) {
        try {
          return (Encoder) method.invoke(null);    
        } catch (IllegalAccessException e) {     
          throw new SssException(e);           
        } catch (InvocationTargetException e) {  
          throw new SssException(e);           
        }
      }
    }
    catch (NoSuchMethodException e) {   
      // When method does not exist, use default.
    }
    return new Encoder() {
      public byte[] encode(Packable p) throws IOException {
        try {
          return ((SelfPackable)p).toBytes();
        }
        catch (Exception e) {
          throw new IOException(e);
        }
      }
    };
  }

  @Override
  public Decoder createDecoder(final Class<? extends Packable> klass) throws SssException {
    assert canEncode(klass): "invalid type";
    try {
      final Method method = klass.getMethod("createDecoder");
      if (method != null && Modifier.isStatic(method.getModifiers())) {
        try {
          return (Decoder) method.invoke(null);    
        } catch (IllegalAccessException e) {     
          throw new SssException(e);           
        } catch (InvocationTargetException e) {  
          throw new SssException(e);           
        }
      }
    }
    catch (NoSuchMethodException e) {
      // When method does not exist, use default.
    }
    try {
      final Constructor<? extends Packable> c = klass.getConstructor();
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          try {
            SelfPackable sp = (SelfPackable)c.newInstance();
            sp.loadBytes(b);
            return sp;
          }
          catch (Exception e) {
            throw new IOException(e);
          }
        }
      };
    }
    catch (NoSuchMethodException e) {
      throw new SssException(klass.getName() + " does not have default constructor.");
    }
  }

  public static SelfEncoding getInstance() {
    return instance;
  }

  private static SelfEncoding instance = new SelfEncoding();
}

